package com.atguigu.flink.cep;

import akka.protobuf.ByteString;
import com.atguigu.flink.function.WaterSensorMapFunction;
import com.atguigu.flink.pojo.WaterSensor;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.cep.CEP;
import org.apache.flink.cep.PatternSelectFunction;
import org.apache.flink.cep.PatternStream;
import org.apache.flink.cep.PatternTimeoutFunction;
import org.apache.flink.cep.pattern.Pattern;
import org.apache.flink.cep.pattern.conditions.SimpleCondition;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.OutputTag;

import java.util.List;
import java.util.Map;

/**
 * Created by Smexy on 2022/12/20
 *
 *      迟到：  当前数据对eventtime  <  watermark，称为迟到。可以使用侧流接收
 *
 *      超时：   指定当前数据对下一条数据，必须在一个规定对时间内到达。
 *                      使用两条数据的 eventtime的间隔  和 时间范围对比
 *
 *                s1,1000,20  指定下一条数据必须在2s只能达到
 *                s2,4000,30
 *                      称 s1为超时数据！
 *
 *    ---------------------------------
 *      对于每一条数据有以下情况:
 *          假设时效范围是2s, 需要要求下一条数据必须是s1
 *              没有超时:
 *                  s1,1000,20
 *                  s2,2000,30   没超时，不会匹配成功。 获取不到
 *
 *
 *                  ------
 *                  s1,1000,20
 *                  s1,2000,20  没超时，会匹配成功。 可以获取到
 *
 *
 *
 *
 *              超时:
 *                   s1,1000,20   超时数据，一定匹配不成功！   可以获取到
 *                   s2,4000,30
 *
 */
public class Demo10_TimeOut
{
    public static void main(String[] args) {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        WatermarkStrategy<WaterSensor> watermarkStrategy = WatermarkStrategy
            .<WaterSensor>forMonotonousTimestamps()
            .withTimestampAssigner((e, r) -> e.getTs());


        SingleOutputStreamOperator<WaterSensor> ds =
            env.readTextFile("data/ceptimeout.txt")
               .map(new WaterSensorMapFunction())
               .assignTimestampsAndWatermarks(watermarkStrategy);


        Pattern<WaterSensor, WaterSensor> pattern = Pattern.<WaterSensor>begin("s1")
            .where(new SimpleCondition<WaterSensor>()
            {
                //定义匹配规则
                @Override
                public boolean filter(WaterSensor value) throws Exception {
                    return "s1".equals(value.getId());
                }
            })
            //下一条必须在2s内到达，不含2s
            .within(Time.seconds(2))
            .next("s2")
            .where(new SimpleCondition<WaterSensor>()
            {
                //定义匹配规则
                @Override
                public boolean filter(WaterSensor value) throws Exception {
                    return "s2".equals(value.getId());
                }
            })
            ;


        PatternStream<WaterSensor> patternStream = CEP.pattern(ds, pattern);

        OutputTag<String> outputTag = new OutputTag<>("timeout", TypeInformation.of(String.class));


        SingleOutputStreamOperator<String> result = patternStream.select(
            outputTag,
            new PatternTimeoutFunction<WaterSensor, String>()
            {
                // Map<String, List<WaterSensor>> pattern: 匹配的是超时对数据
                @Override
                public String timeout(Map<String, List<WaterSensor>> pattern, long timeoutTimestamp) throws Exception {
                    return pattern.toString();
                }
            },
            new PatternSelectFunction<WaterSensor, String>()
            {

                //获取匹配成功的
                @Override
                public String select(Map<String, List<WaterSensor>> map) throws Exception {
                    return map.toString();
                }
            });

        result.getSideOutput(outputTag).printToErr("超时");
        //匹配成功，未超时
        result
                     .print().setParallelism(1);


        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }


    }
}
